package org.apache.zookeeper.server;

import java.util.*;
import java.io.IOException;

import org.apache.zookeeper.proto.PrepareDataRequest;
import org.apache.jute.OutputArchive;
import org.apache.jute.InputArchive;
import org.apache.zookeeper.server.paxos2.*;

public class PrepareDataCache {
    final private static HashMap<Long, TreeMap<Integer, PrepareDataRequest>> cache = new HashMap<Long, TreeMap<Integer, PrepareDataRequest>>();

    public static int size = 0;

    public static void put(long sessionId, int xid, PrepareDataRequest request) {
        if (Configuration.debug)
            System.out.println("Put " + sessionId + " " + xid + " into cache at "+System.currentTimeMillis());
        //System.out.println("buffer ="+request.getData()[0]+""+request.getData()[1]+""+request.getData()[2]+""+request.getData()[3]);
        synchronized (cache) {
            if (!cache.containsKey(sessionId))
                cache.put(sessionId, new TreeMap<Integer, PrepareDataRequest>());
            TreeMap<Integer, PrepareDataRequest> tmp = cache.get(sessionId);
	    if(!tmp.containsKey(xid)){
            	tmp.put(xid, request);
            	size++;
	    }
	    else
		Configuration.decreaseOutstanding();
	    if(size>500)
		System.out.println("cache size="+size);
	    //if(cache.size()>0 && size / cache.size() > Configuration.clientOutStanding * 3 / 2)
	//	System.out.println("prepare size="+size+" outstanding="+Configuration.outstandingCount);
        }
    }

    public static PrepareDataRequest get(long sessionId, int xid) {
        synchronized (cache) {
            if (!cache.containsKey(sessionId) || !cache.get(sessionId).containsKey(xid)) {
                return null;
            } else {
                if (Configuration.debug)
                    System.out.println("Remove " + sessionId + " " + xid + " from cache");
                PrepareDataRequest ret = null;
		TreeMap<Integer, PrepareDataRequest> haha = cache.get(sessionId);
                ret = haha.remove(xid);
                if (ret == null)
                    throw new RuntimeException("weird thing here");
                //System.out.println("cache size "+sessionId+" "+cache.get(sessionId).size());
                size--;

	      	SortedMap tmp = haha.headMap(xid);
		for(int i=0; i<tmp.size(); i++){
			Configuration.decreaseOutstanding();
                        size--;
		}
		tmp.clear();
		/*for(Object old:tmp.keySet()){
			System.out.println(sessionId+" "+xid+" removes "+old);
		     	haha.remove(old);
			Configuration.decreaseOutstanding();
			size--;
		}*/
		if(cache.size()==0 || size/cache.size()<Configuration.clientOutStanding)
            	    cache.notify();
		
                return ret;
            }
        }
    }

    public static void waitNotFull() {
        synchronized (cache) {
	    
            while (cache.size()>0 && size / cache.size() > Configuration.clientOutStanding * 3 / 2){
		System.out.println("prepare cache size = "+size);
                try{
                    cache.wait();
                }
                catch(InterruptedException e){
                    e.printStackTrace();
                }
	    }
        }
    }

    public static void serialize(OutputArchive oa, String tag) throws IOException {
        synchronized (cache) {
            oa.writeInt(cache.size(), "size");
            for (Long sessionId : cache.keySet()) {
                oa.writeLong(sessionId, "sessionId");
                TreeMap<Integer, PrepareDataRequest> client = cache.get(sessionId);
                oa.writeInt(client.size(), "clientsize");
                for (Integer xid : client.keySet()) {
                    oa.writeInt(xid, "xid");
                    oa.writeRecord(client.get(xid), "data");
                    if (Configuration.debug)
                        System.out.println("serialize " + sessionId + " " + xid);
                }
            }
        }
    }

    public static void deserialize(InputArchive ia, String tag) throws IOException {
        synchronized (cache) {
            cache.clear();
            int size = ia.readInt("size");
            System.out.println("deserialize request cache size =" + size);
            for (int i = 0; i < size; i++) {
                long sessionId = ia.readLong("sessionId");
                TreeMap<Integer, PrepareDataRequest> client = new TreeMap<Integer, PrepareDataRequest>();
                int clientSize = ia.readInt("clientsize");
                System.out.println("deserialize client " + sessionId + " " + clientSize);
                for (int j = 0; j < clientSize; j++) {
                    int xid = ia.readInt("xid");
                    PrepareDataRequest req = new PrepareDataRequest();
                    ia.readRecord(req, "data");
                    client.put(xid, req);
                    if (Configuration.debug)
                        System.out.println("deserialize " + sessionId + " " + xid);
                }

                cache.put(sessionId, client);
            }
        }
    }

}
